package com.dx.gate.provider;

import com.dx.chat.api.GateMsgBoxService;
import com.dx.gate.server.ChatServerLauncher;
import com.dx.gate.util.MessageHelper;
import com.dx.mobileimsdk.server.network.MBObserver;
import com.dx.mobileimsdk.server.protocal.Protocal;
import com.dx.mobileimsdk.server.utils.GlobalSendHelper;
import com.dx.mobileimsdk.server.utils.ServerToolKits;
import com.google.gson.Gson;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 收件箱数据发送给在线用户
 *
 * @author xufeng
 * @since 14.04.2023
 */
@Slf4j
@DubboService
public class GateMsgBoxServiceImpl implements GateMsgBoxService, Runnable {

    /**
     * 上行消息缓冲队列
     */
    //public static BlockingQueue<ProtoMsg.Message> down2GateQueue;
    public BlockingQueue<String> down2GateQueue = new LinkedBlockingDeque<>();

    /**
     * 收到rpc新消息
     *
     * @param name
     * @return
     */
    @SneakyThrows
    @Override
    public String receiveMsg(String msg) {
        //down2GateQueue.put(name);
        log.info("来自业务模块的消息，{}", msg);
        send(msg);
        return "";
    }

    /**
     * 发送新消息
     *
     * @param name
     * @return
     */
    @Override
    public String sendMsg(String name) {
        return null;
    }

    @Override
    public void run() {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            executorService.execute(() -> {
                while (true) {
                    String message = null;
                    try {
                        message = down2GateQueue.take();
                        send(message);
                        log.info("收到下行消息：{}", message);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    @SneakyThrows
    public String send(String msg) {
        Protocal protocal = ServerToolKits.toProtocal(msg);
        GlobalSendHelper.sendDataS2C(ChatServerLauncher.getInstance().getServerCoreHandler().getBridgeProcessor(), protocal, new MBObserver() {
            @Override
            public void update(boolean sucess, Object extraObj) {
                if(sucess){
                    log.info("跨集群发送成功");
                }else{
                    log.info("跨集群发送失败");
                }
            }
        });
        return "OK";
    }
}
